package com.atguigu.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.CommonUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Comparator;
import java.util.List;

/**
 * Created by RUI on 2021/6/24 17:04
 */
public class DwdLogApp extends BaseApp {

    public static void main(String[] args) {
        new DwdLogApp().init(2001, 1, "DwdLogApp", "DwdLogApp", Constant.TOPIC_ODS_LOG);
    }

    @Override
    public void run(StreamExecutionEnvironment env,DataStreamSource<String> sourceStream) {

        // 1. 识别新老用户   虽然数据中已经有了这个字段, 但是这个字段可能不准, 所以需要做个纠正
        SingleOutputStreamOperator<JSONObject> validatedStream = distinguishNewOrOld(sourceStream);

        // 2. 分流  不同的日志进入到不同的流中  启动日志   页面  曝光日志
        Tuple3<SingleOutputStreamOperator<JSONObject>, DataStream<JSONObject>, DataStream<JSONObject>> streamTuple3 = splitStream(validatedStream);

        // 3. 把不同流中的数据sink到不同的topic中, 就得到了我们的dwd层的数据
        send2Kafka(streamTuple3);
    }

    private void send2Kafka(Tuple3<SingleOutputStreamOperator<JSONObject>, DataStream<JSONObject>, DataStream<JSONObject>> streamTuple3) {

        streamTuple3.f0
            .map(JSONAware::toJSONString)
            .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_START_LOG));

        streamTuple3.f1
            .map(JSONAware::toJSONString)
            .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_PAGE_LOG));

        streamTuple3.f2
            .map(JSONAware::toJSONString)
            .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_DISPLAY_LOG));
    }

    private Tuple3<SingleOutputStreamOperator<JSONObject>, DataStream<JSONObject>, DataStream<JSONObject>> splitStream(SingleOutputStreamOperator<JSONObject> validatedStream) {
        OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("page") {};
        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("displayTag") {};

        // 启动日志: 主流   页面和曝光进入侧输出流
        SingleOutputStreamOperator<JSONObject> stream = validatedStream
            .process(new ProcessFunction<JSONObject, JSONObject>() {
                @Override
                public void processElement(JSONObject input, Context ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject start = input.getJSONObject("start");

                    if (null != start) { //启动日志
                        out.collect(input);
                    } else {
                        //页面日志
                        JSONObject page = input.getJSONObject("page");
                        if (null != page) {
                            ctx.output(pageTag, input);
                        }

                        //曝光日志
                        JSONArray displays = input.getJSONArray("displays");
                        if (null != displays) {

                            for (int i = 0; i < displays.size(); i++) {
                                JSONObject display = displays.getJSONObject(i);
                                // 把一些其他信息插入到display中
                                display.put("ts", input.getLong("ts"));
                                display.put("page_id", input.getJSONObject("page").getString("page_id"));

                                display.putAll(input.getJSONObject("common"));

                                ctx.output(displayTag, display);
                            }

                        }
                    }
                }
            });

        return Tuple3.of(stream,stream.getSideOutput(pageTag),stream.getSideOutput(displayTag));
    }

    //返回纠正后的流数据
    private SingleOutputStreamOperator<JSONObject> distinguishNewOrOld(DataStreamSource<String> sourceStream) {
        /*
            具体实现逻辑:
            1. 考虑的数据的乱序, 采用事件时间
            2. 窗口
                每个用户的人生中的第一个窗口的中才有可能出现会设置成新用户
                    按照事件时间排序, 最小的那个应该是新用户记录, 其他都应该是旧用户

                其他的窗口都应该是旧用户

            3. 如何识别出现某个窗口是否为这个用户的第一个窗口
                使用状态记录
         */

        return sourceStream
            .map(JSON::parseObject)
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((element, recordTimestamp) -> element.getLong("ts")))
            .keyBy(obj->obj.getJSONObject("common").getString("mid"))
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {

                private ValueState<Long> firstVisitState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    firstVisitState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("firstVisitState", Long.class));
                }

                @Override
                public void process(String key, Context ctx, Iterable<JSONObject> elements, Collector<JSONObject> out) throws Exception {
                    // 判断是否为第一个窗口
                    // 是: 按照时间戳排序, 把最小的is_new设置为1其他设置为0
                    // 不是: 所有的数据的si_new都设置为0
                    if (firstVisitState.value() == null) {

                        List<JSONObject> jsonObjectList = CommonUtil.toList(elements);
                        jsonObjectList.sort(Comparator.comparing(o -> o.getLong("ts")));  // 升序
                        for (int i = 0; i < jsonObjectList.size(); i++) {
                            JSONObject common = jsonObjectList.get(i).getJSONObject("common");
                            if (i==0){
                                common.put("is_new",1);
                                firstVisitState.update(jsonObjectList.get(i).getLong("ts"));
                            }else {
                                common.put("is_new",0);
                            }
                            out.collect(jsonObjectList.get(i));
                        }

                    }else {
                        for (JSONObject obj : elements) {
                            JSONObject common = obj.getJSONObject("common");
                            common.put("is_new", 0);
                            out.collect(obj);
                        }
                    }

                }
            });
    }
}
